//! # 事件总线实现
//!
//! 基于消息总线的事件总线实现，提供完整的事件驱动架构支持

use async_trait::async_trait;
use std::sync::Arc;
use tokio::sync::RwLock;

use crate::bus::MessageBus;
use crate::error::{MessageBusError, MessageBusResult};
use crate::event::{Event, EventBus, EventHandler, EventRouter, EventStats};

/// 基于消息总线的事件总线实现
pub struct MessageBusEventBus {
    message_bus: Arc<dyn MessageBus>,
    event_router: Arc<EventRouter>,
    subscriptions: Arc<RwLock<Vec<String>>>, // 跟踪消息总线订阅ID
}

impl MessageBusEventBus {
    /// 创建新的事件总线
    pub async fn new(message_bus: Arc<dyn MessageBus>) -> MessageBusResult<Self> {
        tracing::info!("正在创建事件总线实例");
        
        let event_router = Arc::new(EventRouter::new());
        let subscriptions = Arc::new(RwLock::new(Vec::new()));

        // 订阅所有事件主题（使用多级通配符）
        tracing::debug!("订阅事件主题: events.**");
        let subscriber = message_bus.subscribe("events.**").await?;
        let subscription_id = subscriber.subscription_id().to_string();
        
        tracing::info!("事件总线成功订阅消息总线: subscription_id={}", subscription_id);
        
        {
            let mut subs = subscriptions.write().await;
            subs.push(subscription_id);
        }

        // 启动消息处理任务
        let router = event_router.clone();
        tracing::debug!("启动事件消息处理任务");
        tokio::spawn(async move {
            let mut message_count = 0u64;
            while let Ok(Some(message)) = subscriber.receive().await {
                message_count += 1;
                tracing::trace!("接收到事件消息 #{}: topic={}", message_count, message.topic());
                
                match Event::from_message(&message) {
                    Ok(event) => {
                        tracing::debug!("解析事件成功: name={}, id={}", event.name, event.id);
                        if let Err(e) = router.route_event(&event).await {
                            tracing::error!("事件路由失败: name={}, id={}, error={}", event.name, event.id, e);
                        } else {
                            tracing::trace!("事件路由成功: name={}, id={}", event.name, event.id);
                        }
                    }
                    Err(e) => {
                        tracing::warn!("无法解析事件消息: topic={}, error={}", message.topic(), e);
                    }
                }
            }
            tracing::warn!("事件消息处理任务结束，处理了 {} 条消息", message_count);
        });

        tracing::info!("事件总线创建完成");

        Ok(Self {
            message_bus,
            event_router,
            subscriptions,
        })
    }

    /// 注册事件处理器
    pub async fn register_handler(&self, handler: Arc<dyn EventHandler>) -> MessageBusResult<String> {
        let handler_id = self.event_router.register_handler(handler.clone()).await?;

        // 为处理器支持的事件自动订阅
        for event_pattern in handler.supported_events() {
            self.event_router.subscribe(&event_pattern, &handler_id).await?;
        }

        Ok(handler_id)
    }

    /// 注销事件处理器
    pub async fn unregister_handler(&self, handler_id: &str) -> MessageBusResult<()> {
        self.event_router.unregister_handler(handler_id).await
    }

    /// 手动订阅事件模式
    pub async fn subscribe_pattern(&self, pattern: &str, handler_id: &str) -> MessageBusResult<()> {
        self.event_router.subscribe(pattern, handler_id).await
    }

    /// 取消事件模式订阅
    pub async fn unsubscribe_pattern(&self, pattern: &str, handler_id: &str) -> MessageBusResult<()> {
        self.event_router.unsubscribe(pattern, handler_id).await
    }

    /// 获取事件路由器（用于高级操作）
    pub fn router(&self) -> &EventRouter {
        &self.event_router
    }
}

#[async_trait]
impl EventBus for MessageBusEventBus {
    async fn publish_event(&self, event: Event) -> MessageBusResult<()> {
        let start_time = std::time::Instant::now();
        tracing::debug!(
            "开始发布事件: name={}, id={}, source_service={}, version={}", 
            event.name, 
            event.id, 
            event.context.source_service, 
            event.version
        );
        
        // 转换事件为消息并发布
        let message = match event.to_message() {
            Ok(msg) => {
                tracing::trace!("事件转换为消息成功: event_id={}, topic={}", event.id, msg.topic());
                msg
            }
            Err(e) => {
                tracing::error!("事件转换为消息失败: event_id={}, error={}", event.id, e);
                return Err(e);
            }
        };
        
        match self.message_bus.publish(message).await {
            Ok(()) => {
                let duration = start_time.elapsed();
                tracing::info!(
                    "事件发布成功: name={}, id={}, duration={:?}", 
                    event.name, 
                    event.id, 
                    duration
                );
                
                // 更新统计信息
                self.event_router.update_published_stats().await;
                Ok(())
            }
            Err(e) => {
                tracing::error!(
                    "事件发布失败: name={}, id={}, error={}", 
                    event.name, 
                    event.id, 
                    e
                );
                Err(e)
            }
        }
    }

    async fn subscribe_event(&self, pattern: &str, handler: Arc<dyn EventHandler>) -> MessageBusResult<String> {
        let handler_id = self.event_router.register_handler(handler).await?;
        self.event_router.subscribe(pattern, &handler_id).await?;
        Ok(handler_id)
    }

    async fn unsubscribe_event(&self, subscription_id: &str) -> MessageBusResult<()> {
        self.event_router.unregister_handler(subscription_id).await
    }

    async fn publish_events(&self, events: Vec<Event>) -> MessageBusResult<()> {
        let mut published_count = 0;
        let mut failed_count = 0;

        // 并行发布事件
        let tasks: Vec<_> = events.into_iter().map(|event| {
            let message_bus = self.message_bus.clone();
            tokio::spawn(async move {
                let message = event.to_message()?;
                message_bus.publish(message).await
            })
        }).collect();

        for task in tasks {
            match task.await {
                Ok(Ok(())) => published_count += 1,
                Ok(Err(_)) | Err(_) => failed_count += 1,
            }
        }

        // 更新统计信息
        for _ in 0..published_count {
            self.event_router.update_published_stats().await;
        }

        if failed_count > 0 {
            tracing::warn!("批量发布事件部分失败: 成功 {}, 失败 {}", published_count, failed_count);
            Err(MessageBusError::internal_error(format!(
                "批量发布事件部分失败: 成功 {}, 失败 {}", published_count, failed_count
            )))
        } else {
            tracing::debug!("批量发布事件成功: {} 个事件", published_count);
            Ok(())
        }
    }

    async fn get_event_stats(&self) -> MessageBusResult<EventStats> {
        Ok(self.event_router.get_stats().await)
    }
}

impl Drop for MessageBusEventBus {
    fn drop(&mut self) {
        // 清理资源
        let message_bus = self.message_bus.clone();
        let subscriptions = self.subscriptions.clone();
        
        tokio::spawn(async move {
            let subs = subscriptions.read().await;
            for subscription_id in subs.iter() {
                if let Err(e) = message_bus.unsubscribe(subscription_id).await {
                    tracing::warn!("清理事件总线订阅失败: {}", e);
                }
            }
        });
    }
}

/// 简单事件处理器实现
pub struct SimpleEventHandler {
    name: String,
    events: Vec<String>,
    handler_fn: Arc<dyn Fn(&Event) -> MessageBusResult<()> + Send + Sync>,
}

impl SimpleEventHandler {
    /// 创建简单事件处理器
    pub fn new<F>(name: String, events: Vec<String>, handler_fn: F) -> Self 
    where
        F: Fn(&Event) -> MessageBusResult<()> + Send + Sync + 'static,
    {
        Self {
            name,
            events,
            handler_fn: Arc::new(handler_fn),
        }
    }
}

#[async_trait]
impl EventHandler for SimpleEventHandler {
    async fn handle(&self, event: &Event) -> MessageBusResult<()> {
        (self.handler_fn)(event)
    }

    fn name(&self) -> &str {
        &self.name
    }

    fn supported_events(&self) -> Vec<String> {
        self.events.clone()
    }
}

/// 异步事件处理器实现
pub struct AsyncEventHandler {
    name: String,
    events: Vec<String>,
    handler_fn: Arc<dyn Fn(&Event) -> std::pin::Pin<Box<dyn std::future::Future<Output = MessageBusResult<()>> + Send>> + Send + Sync>,
}

impl AsyncEventHandler {
    /// 创建异步事件处理器
    pub fn new<F, Fut>(name: String, events: Vec<String>, handler_fn: F) -> Self 
    where
        F: Fn(&Event) -> Fut + Send + Sync + 'static,
        Fut: std::future::Future<Output = MessageBusResult<()>> + Send + 'static,
    {
        let async_fn = Arc::new(move |event: &Event| {
            let fut = handler_fn(event);
            Box::pin(fut) as std::pin::Pin<Box<dyn std::future::Future<Output = MessageBusResult<()>> + Send>>
        });

        Self {
            name,
            events,
            handler_fn: async_fn,
        }
    }
}

#[async_trait]
impl EventHandler for AsyncEventHandler {
    async fn handle(&self, event: &Event) -> MessageBusResult<()> {
        (self.handler_fn)(event).await
    }

    fn name(&self) -> &str {
        &self.name
    }

    fn supported_events(&self) -> Vec<String> {
        self.events.clone()
    }
}

/// 事件聚合器 - 用于收集和聚合相关事件
pub struct EventAggregator {
    name: String,
    aggregate_window_ms: u64,
    max_events: usize,
    collected_events: Arc<tokio::sync::Mutex<Vec<Event>>>,
    last_aggregate_time: Arc<tokio::sync::Mutex<std::time::Instant>>,
    aggregation_fn: Arc<dyn Fn(Vec<Event>) -> MessageBusResult<Option<Event>> + Send + Sync>,
}

impl EventAggregator {
    /// 创建事件聚合器
    pub fn new<F>(
        name: String,
        aggregate_window_ms: u64,
        max_events: usize,
        aggregation_fn: F,
    ) -> Self 
    where
        F: Fn(Vec<Event>) -> MessageBusResult<Option<Event>> + Send + Sync + 'static,
    {
        Self {
            name,
            aggregate_window_ms,
            max_events,
            collected_events: Arc::new(tokio::sync::Mutex::new(Vec::new())),
            last_aggregate_time: Arc::new(tokio::sync::Mutex::new(std::time::Instant::now())),
            aggregation_fn: Arc::new(aggregation_fn),
        }
    }

    /// 添加事件到聚合器
    pub async fn add_event(&self, event: Event) -> MessageBusResult<Option<Event>> {
        let mut events = self.collected_events.lock().await;
        let mut last_time = self.last_aggregate_time.lock().await;

        events.push(event);

        // 检查是否需要触发聚合
        let should_aggregate = events.len() >= self.max_events ||
            last_time.elapsed().as_millis() as u64 >= self.aggregate_window_ms;

        if should_aggregate {
            let collected = std::mem::take(&mut *events);
            *last_time = std::time::Instant::now();
            drop(events);
            drop(last_time);

            if !collected.is_empty() {
                return (self.aggregation_fn)(collected);
            }
        }

        Ok(None)
    }

    /// 强制聚合当前收集的事件
    pub async fn force_aggregate(&self) -> MessageBusResult<Option<Event>> {
        let mut events = self.collected_events.lock().await;
        let mut last_time = self.last_aggregate_time.lock().await;

        if events.is_empty() {
            return Ok(None);
        }

        let collected = std::mem::take(&mut *events);
        *last_time = std::time::Instant::now();
        drop(events);
        drop(last_time);

        (self.aggregation_fn)(collected)
    }
}

#[async_trait]
impl EventHandler for EventAggregator {
    async fn handle(&self, event: &Event) -> MessageBusResult<()> {
        let _aggregated_event = self.add_event(event.clone()).await?;
        // 这里可以选择是否发布聚合后的事件
        Ok(())
    }

    fn name(&self) -> &str {
        &self.name
    }

    fn supported_events(&self) -> Vec<String> {
        vec!["*".to_string()] // 默认处理所有事件
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::event::{Event, EventContext, EventType};
    use crate::memory::InMemoryMessageBus;
    use serde_json::json;
    use std::sync::atomic::{AtomicU32, Ordering};

    #[tokio::test]
    async fn test_event_bus_creation() {
        let message_bus = Arc::new(InMemoryMessageBus::new().await.unwrap());
        let event_bus = MessageBusEventBus::new(message_bus).await.unwrap();
        
        let stats = event_bus.get_event_stats().await.unwrap();
        assert_eq!(stats.total_published, 0);
        assert_eq!(stats.active_handlers, 0);
    }

    #[tokio::test]
    async fn test_event_publish_and_handle() {
        let message_bus = Arc::new(InMemoryMessageBus::new().await.unwrap());
        let event_bus = MessageBusEventBus::new(message_bus).await.unwrap();

        // 给事件总线一些时间初始化
        tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;

        // 创建计数器
        let counter = Arc::new(AtomicU32::new(0));
        let counter_clone = counter.clone();

        // 创建事件处理器
        let handler = Arc::new(SimpleEventHandler::new(
            "test-handler".to_string(),
            vec!["user.created".to_string()],
            move |event| {
                counter_clone.fetch_add(1, Ordering::Relaxed);
                println!("处理事件: {}", event.name);
                Ok(())
            },
        ));

        // 注册处理器
        let _handler_id = event_bus.register_handler(handler).await.unwrap();

        // 等待处理器注册完成
        tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;

        // 创建和发布事件
        let context = EventContext::new("test-service".to_string());
        let event = Event::new(
            "user.created".to_string(),
            EventType::Domain,
            json!({"user_id": "123", "email": "test@example.com"}),
            context,
        );

        event_bus.publish_event(event).await.unwrap();

        // 等待事件处理（增加等待时间）
        tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;

        // 验证处理器被调用
        let final_count = counter.load(Ordering::Relaxed);
        println!("最终计数器值: {}", final_count);
        assert_eq!(final_count, 1);

        let stats = event_bus.get_event_stats().await.unwrap();
        println!("事件统计: 已发布={}, 已处理={}", stats.total_published, stats.total_processed);
        assert_eq!(stats.total_published, 1);
        // 不检查 total_processed，因为这个值可能由于异步处理延迟更新
    }

    #[tokio::test]
    async fn test_batch_event_publish() {
        let message_bus = Arc::new(InMemoryMessageBus::new().await.unwrap());
        let event_bus = MessageBusEventBus::new(message_bus).await.unwrap();

        let context = EventContext::new("test-service".to_string());
        let events = vec![
            Event::new("event1".to_string(), EventType::Domain, json!({"id": 1}), context.clone()),
            Event::new("event2".to_string(), EventType::Domain, json!({"id": 2}), context.clone()),
            Event::new("event3".to_string(), EventType::Domain, json!({"id": 3}), context),
        ];

        event_bus.publish_events(events).await.unwrap();

        let stats = event_bus.get_event_stats().await.unwrap();
        assert_eq!(stats.total_published, 3);
    }

    #[tokio::test]
    async fn test_event_aggregator() {
        let aggregator = EventAggregator::new(
            "test-aggregator".to_string(),
            1000, // 1秒聚合窗口
            3,    // 最多3个事件
            |events| {
                if events.is_empty() {
                    return Ok(None);
                }

                let context = EventContext::new("aggregator".to_string());
                let aggregated_event = Event::new(
                    "events.aggregated".to_string(),
                    EventType::System,
                    json!({"count": events.len(), "events": events.len()}),
                    context,
                );
                Ok(Some(aggregated_event))
            },
        );

        let context = EventContext::new("test-service".to_string());
        
        // 添加事件但不应该触发聚合
        let event1 = Event::new("test1".to_string(), EventType::Domain, json!({}), context.clone());
        let result1 = aggregator.add_event(event1).await.unwrap();
        assert!(result1.is_none());

        let event2 = Event::new("test2".to_string(), EventType::Domain, json!({}), context.clone());
        let result2 = aggregator.add_event(event2).await.unwrap();
        assert!(result2.is_none());

        // 第三个事件应该触发聚合
        let event3 = Event::new("test3".to_string(), EventType::Domain, json!({}), context);
        let result3 = aggregator.add_event(event3).await.unwrap();
        assert!(result3.is_some());
        
        let aggregated = result3.unwrap();
        assert_eq!(aggregated.name, "events.aggregated");
    }
}